package org.databandtech.job.utils;

import java.util.Collection;
import java.util.List;
import java.util.ResourceBundle;

import org.apache.sqoop.client.SqoopClient;
import org.apache.sqoop.model.MConfig;
import org.apache.sqoop.model.MConnector;
import org.apache.sqoop.model.MInput;
import org.apache.sqoop.model.MJob;
import org.apache.sqoop.model.MLink;
import org.apache.sqoop.model.MLinkConfig;
import org.apache.sqoop.validation.Status;

public class SqoopUtils {

	public static void showAllConnectors(SqoopClient client) {
		Collection<MConnector> connectors = client.getConnectors();

		for (MConnector conn : connectors) {
			System.out.println(conn.getUniqueName() + " : " + conn.getClassName());
		}
	}

	public static void showAllLinks(SqoopClient client) {
		List<MLink> links = client.getLinks();
		for (MLink link : links) {
			System.out.println("Link Name : " + link.getName());
		}
	}

	public static void showAllJobs(SqoopClient client) {
		List<MJob> jobs = client.getJobs();
		for (MJob job : jobs) {
			System.out.println("Job Name : " + job.getName());
		}
	}

	public static void displayConfig(SqoopClient client, String connectorName) {
		// link config
		describe(client.getConnector(connectorName).getLinkConfig().getConfigs(),
				client.getConnectorConfigBundle(connectorName));
		// from job config
		describe(client.getConnector(connectorName).getFromConfig().getConfigs(),
				client.getConnectorConfigBundle(connectorName));
		// to job config
		describe(client.getConnector(connectorName).getToConfig().getConfigs(),
				client.getConnectorConfigBundle(connectorName));
	}

	public static void deleteLink(final SqoopClient client, final String linkname) {
		client.deleteLink(linkname);
	}

	public static void createjdbcLink(SqoopClient client, String linkname, String connection, String user, String jdbcuser,
			String jdbcpassword) {
		MLink link = client.createLink("generic-jdbc-connector");
		link.setName(linkname);
		link.setCreationUser(user);
		MLinkConfig linkConfig = link.getConnectorLinkConfig();
		// config values
		linkConfig.getStringInput("linkConfig.connectionString").setValue(connection);
		linkConfig.getStringInput("linkConfig.jdbcDriver").setValue("com.mysql.jdbc.Driver");
		linkConfig.getStringInput("linkConfig.username").setValue(jdbcuser);
		linkConfig.getStringInput("linkConfig.password").setValue(jdbcpassword);
		linkConfig.getStringInput("linkConfig.password").setValue(jdbcpassword);
		linkConfig.getStringInput("dialect.identifierEnclose").setValue(" ");
		Status status = client.saveLink(link);
		if (status.canProceed()) {
			System.out.println("Created Link with Link Name : " + link.getName());
		} else {
			System.out.println("Something went wrong creating the link");
		}
	}

	public static void createhdfsLink(SqoopClient client, String linkname, String user, String uri, String confDir) {

		MLink link = client.createLink("hdfs-connector");
		link.setName(linkname);
		link.setCreationUser(user);
		MLinkConfig linkConfig = link.getConnectorLinkConfig();
		// config values
		linkConfig.getStringInput("linkConfig.uri").setValue(uri);
		linkConfig.getStringInput("linkConfig.confDir").setValue(confDir);
		Status status = client.saveLink(link);
		if (status.canProceed()) {
			System.out.println("Created Link with Link Name : " + link.getName());
		} else {
			System.out.println("Something went wrong creating the link");
		}

	}

	public static void describe(List<MConfig> configs, ResourceBundle resource) {
		for (MConfig config : configs) {
			System.out.println(resource.getString(config.getLabelKey()) + ":");
			List<MInput<?>> inputs = config.getInputs();
			for (MInput<?> input : inputs) {
				System.out.println(
						resource.getString(input.getLabelKey()) + " <" + input.getName() + "> " + input.getValue());
			}
			System.out.println();
		}
	}

}